package com.shujia.filter;

import com.shujia.been.Student;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

public class Code07MoreFilter {
    static Connection connection;

    static {
        // 创建连接需要给定连接信息，所以需要给定配置类
        Configuration conf = new Configuration();

        // 由于对HBASE进行操作时，需要连接其Zookeeper 那么需要传入Zookeeper的IP
        conf.set("hbase.zookeeper.quorum", "node1,node2,master");
        // 获取连接对象 通过ConnectionFactory来创建连接对象

        try {
            connection = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static boolean tableExists(Admin admin, TableName tableName) {
        boolean exists = false;
        try {
            exists = admin.tableExists(tableName);
            if (exists) {
                System.out.println("表存在..");
            } else {
                System.out.println("表不存在");
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
        return exists;
    }


    /**
     * 需求：针对学生表数据 需要过滤以下条件数据：
     *      1.过滤ID大于15001001的学生
     *      2.过滤班级为文科班的学生
     *      3.过滤年龄大于22的学生
     *      4.过滤性别为女的学生
      */

    public static void main(String[] args) throws IOException {

        Admin admin = connection.getAdmin();
        TableName tableName = TableName.valueOf("api:tb1");

        if (tableExists(admin, tableName)) {
            Table table = connection.getTable(tableName);

            Scan scan = new Scan();

            // 由于过滤条件较多，需要通过多个过滤器来实现过滤，所以需要创建FilterList
            FilterList filterList = new FilterList();



//            1.过滤ID大于15001001的学生
            RowFilter rowFilter = new RowFilter(
                    CompareFilter.CompareOp.GREATER,
                    new BinaryComparator(Bytes.toBytes("15001001"))
            );

            // 可以添加过滤器
            filterList.addFilter(rowFilter);


            // 2.过滤班级为文科班的学生

            /**
             *  public SingleColumnValueFilter(final byte [] family, final byte [] qualifier,
             *       final CompareOp compareOp, final ByteArrayComparable comparator)
             */
            // 在单列值过滤器中，可以对数值进行判断，也可以对字符串进行比较，比较方式如下：
            //  除了使用SubstringComparator 还可以使用 RegexStringComparator 比较字符串
            SingleColumnValueFilter clazzFilter = new SingleColumnValueFilter(
                    Bytes.toBytes("info2"),
                    Bytes.toBytes("clazz"),
                    CompareFilter.CompareOp.EQUAL,
                    new SubstringComparator("文科")
            );

            filterList.addFilter(clazzFilter);

            // 3.过滤年龄大于22的学生

            SingleColumnValueFilter ageFilter = new SingleColumnValueFilter(
                    Bytes.toBytes("info1"),
                    Bytes.toBytes("age"),
                    CompareFilter.CompareOp.GREATER,
                    Bytes.toBytes(22)
            );

            filterList.addFilter(ageFilter);

            // 4.过滤性别为女的学生
            SingleColumnValueFilter genderFilter = new SingleColumnValueFilter(
                    Bytes.toBytes("info1"),
                    Bytes.toBytes("gender"),
                    CompareFilter.CompareOp.EQUAL,
                    Bytes.toBytes("女")
            );
            filterList.addFilter(genderFilter);



            // 将合并的Filter放置scan中获取数据
            scan.setFilter(filterList);

            ResultScanner scanner = table.getScanner(scan);

            printData(scanner);

            admin.close();
            table.close();
            connection.close();

        }


    }

    public static void printData(ResultScanner scanner) {
        for (Result result : scanner) {
            String rowKey = Bytes.toString(result.getRow());

            String name = Bytes.toString(result.getValue("info1".getBytes(StandardCharsets.UTF_8)
                    , "name".getBytes(StandardCharsets.UTF_8)));

            int age = Bytes.toInt(result.getValue("info1".getBytes(StandardCharsets.UTF_8)
                    , "age".getBytes(StandardCharsets.UTF_8)));

            String gender = Bytes.toString(result.getValue("info1".getBytes(StandardCharsets.UTF_8)
                    , "gender".getBytes(StandardCharsets.UTF_8)));

            String clazz = Bytes.toString(result.getValue("info2".getBytes(StandardCharsets.UTF_8)
                    , "clazz".getBytes(StandardCharsets.UTF_8)));

            Student student = new Student(rowKey, name, age, gender, clazz);
            System.out.println(student);

//            System.out.println("name:"+name+" age:"+age+" gender:"+gender+" clazz:"+clazz);

        }
    }
}
